package com.icarbonx.dap.notification.job

import com.google.gson.Gson
import com.icarbonx.dap.notification.msgservice.rabbitmq.{ConnUtil, MQObject}
import com.icarbonx.dap.notification.util._
import com.icarbonx.dap.notification.vo._
import kafka.serializer.StringDecoder
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}

import scala.collection.mutable.ArrayBuffer

/**
 * Created by Lenovo on 2018/8/12.
 * 6:00&前后两个点血糖>7.8mmol/L，及3:00 前后两个点血糖<3.3mmol/L&胰岛素/口服降糖药治疗的用户
 */
object BloodSugarSomogyiStream {
  def main(args: Array[String]) {
    Logger.getLogger("org.apache.kafka").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.zookeeper").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    /**
     * brokers = "10.47.250.4:9092,10.47.250.7:9092,10.47.250.6:9092"
     * topicList = "abel_test"
     * mq:10.104.103.7 5672 guest icx@betaOn16 /
     * redis:10.104.103.7 6379 passxx#$#dslllfjshwerw123@ 3000 abel_test
     * brokers = "10.100.240.3:9092,10.100.240.240:9092,10.100.240.2:9092"
     * topicList = "bsmp.blood_glucose"
     * mq:10.0.7.101 5672 guest guest@icx  /
     * redis:172.18.166.122 6379 dev
     */
    println(args.mkString("\t"))
    val Array(brokers, topicList, mqHost, mqPort, mqUsername, mqPassword,
    mqVirtualHost, exchangeName, mqRouteKey, checkpointDir, duration, rddDuration,
    jdbcURL, username, password, appName
    ) = args

    val conf = SparkUtil.getSparkConf(appName, Array(classOf[BloodSugarRecord]))

    val sc = SparkContext.getOrCreate(conf)

    val updateFunc = (values: Seq[ArrayBuffer[BloodSugarRecord]],
                      state: Option[ArrayBuffer[BloodSugarRecord]]) => {
      //保存状态
      val todayTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss") - 7 * 60 * 60 * 1000
      val currentCount = ArrayBuffer[BloodSugarRecord]()
      values.foreach { list => currentCount ++= list }
      val previousCount = state.getOrElse(ArrayBuffer[BloodSugarRecord]())
      Some(currentCount ++ previousCount.filter { record =>
        record.timestamp > todayTimeStamp
      })
    }

    //    val updateDayCountFunc =
    //      (values: Seq[mutable.HashMap[String, Int]],
    //       state: Option[mutable.HashMap[String, Int]]) => {
    //        //保存状态
    //        val fiveDaysAgoTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss") - 5 * 24 * 60 * 60 * 1000
    //        val currentState = mutable.HashMap[String, Int]()
    //        val previousState = state.getOrElse(mutable.HashMap[String, Int]())
    //        values.foreach {
    //          case map =>
    //            map.foreach {
    //              case (timeState, flag) =>
    //                currentState += ((timeState, flag))
    //            }
    //        }
    //        val tempOldState = previousState.filter {
    //          case (timeState, flag) =>
    //            val dayTime = timeState.split(",")(0).toLong
    //            !currentState.contains(timeState) && dayTime > fiveDaysAgoTimeStamp
    //        }
    //        Some(currentState ++= tempOldState)
    //      }

    def creatingFunc(): StreamingContext = {

      val ssc = new StreamingContext(sc, Seconds(duration.toInt))
      ssc.checkpoint(checkpointDir)
      val kafkaParams = KafkaUtil.getKafkaParam(brokers, appName)

      val mqConn = new MQObject(
        mqHost,
        mqPort.toInt,
        mqUsername,
        mqPassword,
        mqVirtualHost
      )

      val topics = topicList.split(",").toSet
      val bloodSugarMessages = KafkaUtils
        .createDirectStream[String, String, StringDecoder, StringDecoder](
          ssc,
          kafkaParams,
          topics
        )

      val newBatchAndOld: DStream[(String, ArrayBuffer[BloodSugarRecord])] =
        bloodSugarMessages
          .filter(_._2.nonEmpty)
          .flatMap { msg =>
            try {
              println("___________ msg :" + msg._2)
              val gson = new Gson()
              val voList =
                gson.fromJson(msg._2, classOf[Array[BloodSugarRecord]])
              voList.map { vo =>
                (vo.personId.toString, ArrayBuffer[BloodSugarRecord](vo))
              }.toList
            } catch {
              case e: Exception =>
                println("____________" + e.getMessage)
                List(("", ArrayBuffer[BloodSugarRecord]()))
            }
          }
          .filter(_._1.nonEmpty)
          .updateStateByKey(updateFunc)

      val notNullRdd = newBatchAndOld.filter(_._2.nonEmpty)

      notNullRdd.checkpoint(Duration(rddDuration.toInt * 1000))

      val sortRdd = notNullRdd.map { case (pid, list) => (pid, list.sortBy(_.timestamp)) }

      val drugRecord = ssc.receiverStream(new MysqlReceiver(jdbcURL, username, password))

      val drugUser = sortRdd.join(drugRecord)

      drugUser.foreachRDD(mess => {
        mess.foreachPartition(lines => {
          //            val nowTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss")
          val mqMsg = ArrayBuffer[String]()
          val gson = new Gson()
          lines.foreach {
            case (pid, (list, drugTime)) =>
              println(s"______ pid ${pid} list  ${list}  drugTime ${drugTime} ")
              //@todo 26 6:00&前后两个点血糖>7.8mmol/L，及3:00 前后两个点血糖<3.3mmol/L&胰岛素/口服降糖药治疗的用户
              //              val userFlag = HttpGetRequest.doGet(apiUrl, pid)
              val rangeThreeList = list.filter { vo =>
                vo.timestamp >= DateUtil.getTodayATime(2, 45) && vo.timestamp <= DateUtil.getTodayATime(3, 15)
              }
              val rangeSixList = list.filter { vo =>
                vo.timestamp >= DateUtil.getTodayATime(5, 45) && vo.timestamp <= DateUtil.getTodayATime(6, 15)
              }

              println(s"______ pid ${pid} rangeThreeList  ${rangeThreeList}  rangeSixList ${rangeSixList} ")
              if (rangeThreeList.size > 1 && rangeSixList.size > 1) {
                println(s"_____ pid $pid ,\t drugTime ${drugTime}   rangeThreeList   ${rangeThreeList.toString()} " +
                  s" rangeSixList  ${rangeSixList.toString()} ")
                // 3点连续连个点
                var resultFlag = 0
                var pre = rangeThreeList.head
                for (i <- 1 to rangeThreeList.size - 1) {
                  val now = rangeThreeList(i)
                  if (pre.value < 3.3 && now.value < 3.3 && now.timestamp - pre.timestamp < 15 * 60 * 60 * 1000) {
                    resultFlag = 1
                  }
                  pre = now
                }
                pre = rangeSixList.head
                for (i <- 1 to rangeSixList.size - 1) {
                  val now = rangeSixList(i)
                  if (pre.value > 7.8 && now.value > 7.8 && now.timestamp - pre.timestamp < 15 * 60 * 60 * 1000) {
                    resultFlag += 1
                  }
                  pre = now
                }
                if (resultFlag > 1) {
                  val msg = new WarningMsg(
                    "glycemia",
                    pid.toLong,
                    NoticeMsg.somogyi._1,
                    NoticeMsg.somogyi._2,
                    NoticeMsg.somogyi._3,
                    rangeThreeList.head.timestamp.toString,
                    NoticeMsg.somogyi._5,
                    NoticeMsg.somogyi._4,
                    NoticeMsg.somogyi._6
                  )
                  mqMsg += gson.toJson(msg)
                }
              }

          }
          if (mqMsg.nonEmpty) {
            val mqChannel = mqConn.connect().createChannel()
            mqMsg.foreach { msg =>
              ConnUtil.sendMsg(mqChannel, exchangeName, mqRouteKey, msg)
            }
            mqChannel.close()
          }
        })

      })

      ssc
    }


    val ssc = StreamingContext.getOrCreate(checkpointDir, creatingFunc)

    ssc.start()
    ssc.awaitTermination()
  }

}